package com.jie.flink.cdc.listener;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * @author zhanggj
 * @data 2023/2/17
 */
@Slf4j
public class KafkaListenerTest {
    private static final Properties KAFKA_PROPERTIES = new Properties(){
        {
            put("bootstrap.servers", "127.0.0.1:9092");
            put("group.id", "kafka_group_id");
            put("enable.auto.commit", "true");
            put("auto.commit.interval.ms", "1000");
            put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
            put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        }
    };

    private static final String TOPIC_NAME = "topic_name";

    public static void main(String[] args) {
        KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(KAFKA_PROPERTIES);
        consumer.subscribe(Arrays.asList(TOPIC_NAME));
        boolean isClose = false;
        while (true) {
            while(!isClose) {
                ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> record : records) {
                    log.info("consumer topic:{}, key={}, value={}", TOPIC_NAME, record.key(), record.value());
                }
            }
            consumer.close();
        }
    }
}
